package cn.learn.delay.core;

import cn.learn.delay.message.BaseRedisMessage;
import com.alibaba.fastjson.JSON;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import utils.json.JsonUtil;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.*;

/**
 * @author : Yu.Tan
 * @date : 2022/10/11 17:13
 * @desc : **
 */
@Slf4j
public abstract class AbstractRedisDelayExecute<MSG extends BaseRedisMessage> implements RedisDelayExecute<MSG> {

    @Autowired
    private RedisDelayTemplate redisDelayTemplate;

    private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    private static final int RANG_SIZE = 10;


    public boolean send(MSG message) {
        Assert.notNull(message, "消息不能为空！");
        if (StringUtils.isBlank(message.getMessageId())) {
            message.setMessageId(UUID.randomUUID().toString().replace("-", ""));
            log.info("消息，messageId为空，生成id={}", message.getMessageId());
        }
        try {
            LocalDateTime delayLocalDateTime = message.getSendTime().plusSeconds(message.getDelayTime());
            double delayTime = delayLocalDateTime.toInstant(ZoneOffset.of("+8")).toEpochMilli();
            redisDelayTemplate.getRedisTemplate().opsForHash().put(MSG_POOL + message.getGroup() + message.getMessageId(), message.getMessageId(), JSON.toJSONString(message));
            redisDelayTemplate.getRedisTemplate().opsForZSet().add(MSG_POOL + message.getGroup(), message.getMessageId(), delayTime);
            log.info("RedisMq发送消费信息{}，当前时间:{},消费时间预计{}", message, new Date(), FORMATTER.format(delayLocalDateTime));
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public <R> List<R> consumerMessage(String group, Class<R> clazz) {
        //条件score 为 当前时间+1秒的毫秒值
        double to = LocalDateTime.now().plusSeconds(1).toInstant(ZoneOffset.of("+8")).toEpochMilli();
        Set<ZSetOperations.TypedTuple<Object>> typedTuples = redisDelayTemplate.getRedisTemplate().opsForZSet()
                .rangeByScoreWithScores(MSG_POOL + group, 0, to, 0, RANG_SIZE);
//        log.info("group={},clazz={},key={},min={},max={},offset={},count={},result={}", group, clazz, MSG_POOL + group, 0, to, 0, RANG_SIZE, typedTuples);
        List<R> results = new ArrayList<>();
        List<String> messageIds = new ArrayList<>();
        if (!CollectionUtils.isEmpty(typedTuples)) {
            for (ZSetOperations.TypedTuple<Object> tuple : typedTuples) {
                String messageId = (String) tuple.getValue();
                messageIds.add(messageId);
                //删除消息队列
                redisDelayTemplate.getRedisTemplate().opsForZSet().remove(MSG_POOL + group, messageId);
                log.info("消费到数据messageId key={} messageId={}", MSG_POOL + group, messageId);
            }
        }
        if (!CollectionUtils.isEmpty(messageIds)) {
            for (String messageId : messageIds) {
                String value = (String) redisDelayTemplate.getRedisTemplate().opsForHash().get(MSG_POOL + group + messageId, messageId);
                log.info("消费到业务数据 key{},messageId={},message={}", MSG_POOL + group + messageId, messageId, value);
                R r = JSON.parseObject(value, clazz);
                results.add(r);
                //删除信息池
                redisDelayTemplate.getRedisTemplate().opsForHash().delete(MSG_POOL + group + messageId, messageId);
                log.info("删除业务业务数据 key={},messageId={}", MSG_POOL + group + messageId, messageId);
            }
        }
        return results;
    }


    public void dealConsumer(RedisDelayExecute<MSG> execute, MSG msg) {
        boolean deal = execute.dealMessage(msg);
        if (deal) {
            return;
        }
        Integer failCount = msg.getFailCount();
        if ( failCount < 3) {
            log.info("消息={}，失败次数={} 小于阈值，重试！", msg, failCount);
            msg.setFailCount(failCount+1);
            send(msg);
        } else {
            log.info("消息={}，失败次数={} 大于阈值，不在重试！", msg, failCount);
        }
    }

}
